// 16_async_programming.rs
// 演示 Rust 中的异步编程(async/await)概念和用法

// 导入异步编程所需的模块
use std::time::Duration;
use tokio::time::sleep;
use tokio::sync::{oneshot, Mutex};
use std::sync::Arc;
use std::error::Error;
use futures::future::join_all;
use futures::stream::{self, StreamExt};
use std::string::String;

// 注意：要运行此示例，需要在 Cargo.toml 中添加以下依赖：
// [dependencies]
// tokio = { version = "1", features = ["full"] }
// futures = "0.3"

// 将原来的 main 函数改为公共的异步 run 函数
pub async fn run() {
    // 1. 基本的异步函数和 await
    println!("=== 基本的异步函数和 await ===");
    
    // 调用异步函数并等待其完成
    let result = hello_world().await;
    println!("异步函数返回: {}", result);
    
    // 2. 并行执行异步任务
    println!("\n=== 并行执行异步任务 ===");
    
    // 使用 join! 宏并行执行多个异步任务
    let (result1, result2) = tokio::join!(
        task_with_delay(2, "任务1"),
        task_with_delay(1, "任务2")
    );
    
    println!("任务1结果: {}", result1);
    println!("任务2结果: {}", result2);
    
    // 3. 动态数量的异步任务
    println!("\n=== 动态数量的异步任务 ===");
    
    let mut tasks = Vec::new();
    for i in 1..=5 {
        let task_name = Arc::new(format!("动态任务{}", i));
        let task_name_clone = task_name.clone();
        tasks.push(async move {
            task_with_delay(i, &task_name_clone).await
        });
    }
    
    let results = join_all(tasks).await;
    for result in results {
        println!("动态任务结果: {}", result);
    }
    
    // 4. 异步错误处理
    println!("\n=== 异步错误处理 ===");
    
    // 成功的异步操作
    if let Ok(result) = maybe_fail(false).await {
        println!("成功操作结果: {}", result);
    }
    
    // 失败的异步操作
    if let Err(err) = maybe_fail(true).await {
        println!("预期的错误: {}", err);
    }
    
    // 5. 异步流处理
    println!("\n=== 异步流处理 ===");
    
    let mut stream = count_up_to(5).await;
    while let Some(count) = stream.next().await {
        println!("流元素: {}", count);
    }
    
    // 6. 异步共享状态
    println!("\n=== 异步共享状态 ===");
    
    let counter = Arc::new(Mutex::new(0));
    let mut handles = Vec::new();
    
    for i in 0..10 {
        let counter = counter.clone();
        handles.push(tokio::spawn(async move {
            increment_counter(counter, i).await;
        }));
    }
    
    // 等待所有任务完成
    for handle in handles {
        handle.await.unwrap();
    }
    
    println!("最终计数器值: {}", *counter.lock().await);
    
    // 7. oneshot 通道通信
    println!("\n=== oneshot 通道通信 ===");
    
    let (tx, rx) = oneshot::channel();
    
    // 发送方任务
    tokio::spawn(async move {
        sleep(Duration::from_secs(1)).await;
        tx.send("消息已送达").unwrap();
    });
    
    // 接收方等待消息
    match rx.await {
        Ok(message) => println!("接收到消息: {}", message),
        Err(_) => println!("发送方已关闭"),
    }
    
    // 8. 超时处理
    println!("\n=== 超时处理 ===");
    
    // 不会超时的操作
    let result = tokio::time::timeout(
        Duration::from_secs(3),
        task_with_delay(2, "不会超时的任务")
    ).await;
    
    match result {
        Ok(value) => println!("操作完成: {}", value),
        Err(_) => println!("操作超时"),
    }
    
    // 将会超时的操作
    let result = tokio::time::timeout(
        Duration::from_secs(1),
        task_with_delay(2, "将会超时的任务")
    ).await;
    
    match result {
        Ok(value) => println!("操作完成: {}", value),
        Err(_) => println!("操作超时"),
    }
    
    // 9. 异步块
    println!("\n=== 异步块 ===");
    
    // 直接使用异步块
    let async_block = async {
        sleep(Duration::from_secs(1)).await;
        "异步块执行完成"
    };
    
    println!("异步块结果: {}", async_block.await);
    
    // 10. select! 宏 - 等待多个事件中的第一个完成
    println!("\n=== select! 宏 ===");
    
    tokio::select! {
        result = task_with_delay(1, "select任务1") => {
            println!("任务1先完成: {}", result);
        },
        result = task_with_delay(2, "select任务2") => {
            println!("任务2先完成: {}", result);
        },
        _ = tokio::time::sleep(Duration::from_secs(3)) => {
            println!("超时");
        }
    }
    
    println!("\n异步编程示例演示完成！");
    println!("Rust 的异步编程提供了高效处理并发任务的方式");
    println!("关键概念：async 函数、await 表达式、Future trait、异步运行时");
}

// 基本的异步函数
async fn hello_world() -> String {
    // 模拟异步操作
    sleep(Duration::from_secs(1)).await;
    "Hello, Async World!".to_string()
}

// 带延迟的异步任务
async fn task_with_delay(seconds: u64, name: &str) -> String {
    println!("{} 开始执行，将延迟 {} 秒", name, seconds);
    sleep(Duration::from_secs(seconds)).await;
    format!("{} 执行完成", name)
}

// 可能失败的异步操作
async fn maybe_fail(should_fail: bool) -> Result<String, Box<dyn Error>> {
    sleep(Duration::from_secs(1)).await;
    
    if should_fail {
        Err("操作失败".into())
    } else {
        Ok("操作成功".to_string())
    }
}

// 异步流生成器
async fn count_up_to(max: u32) -> impl futures::Stream<Item = u32> {
    let mut values = Vec::new();
    for i in 1..=max {
        sleep(Duration::from_millis(500)).await;
        values.push(i);
    }
    stream::iter(values)
}

// 异步更新共享计数器
async fn increment_counter(counter: Arc<Mutex<i32>>, id: u32) {
    // 获取锁并更新计数器
    let mut value = counter.lock().await;
    *value += 1;
    println!("任务 {} 将计数器增加到 {}", id, *value);
    
    // 锁会在value离开作用域时自动释放
}